package com.asiainfo.dacp.common.twophasesink;

import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

/**
 * Doris 事务上下文
 */

public class DorisTransaction implements Serializable {
    private static final long serialVersionUID = 1L;
    private final transient Connection connection;
    private final List<String> records;
    private final String tableName;
    private final int batchSize;
    private transient PreparedStatement preparedStatement;
    private int recordCount;

    private  static  final Logger log= LoggerFactory.getLogger(DorisTransaction.class);
    public DorisTransaction() {
        // 用于反序列化时的构造函数
        this.connection = null;
        this.records = new ArrayList<>();
        this.tableName = null;
        this.batchSize = 0;
        this.recordCount = 0;
    }


    public DorisTransaction(Connection connection, List<String> records,
                            String tableName, int batchSize) {
        this.connection = connection;
        this.records = new ArrayList<>(records);
        this.tableName = tableName;
        this.batchSize = batchSize;
        this.recordCount = 0;
        initializePreparedStatement();
    }

    /**
     * 初始化预编译SQL语句
     * 根据表名构建 INSERT 语句并创建 PreparedStatement 对象
     *
     * @throws RuntimeException 当SQL执行异常时抛出运行时异常
     */
    private void initializePreparedStatement() {
        try {
            // 根据实际表结构调整 SQL
            String sql = "INSERT INTO " + tableName + "  VALUES (?)";
            this.preparedStatement = connection.prepareStatement(sql);
        } catch (SQLException e) {
            throw new RuntimeException("初始化 PreparedStatement 失败", e);
        }
    }

    /**
     * 添加一条记录到记录集合中
     *
     * @param record 要添加的记录字符串
     */

    public void addRecord(String record) {
        records.add(record);
        recordCount++;
    }

    /**
     * 判断是否达到批处理阈值需要刷新
     *
     * @return true-需要刷新，false-不需要刷新
     */

    public boolean shouldFlush() {
        return recordCount >= batchSize;
    }

    /**
     * 执行批量写入操作
     * 将缓冲区中的数据通过 PreparedStatement 批量插入到 Doris 表中
     *
     * @throws SQLException SQL执行异常
     */

    public void executeBatch() throws SQLException {
        if (records.isEmpty()) {
            return;
        }

        for (String record : records) {
            // 根据实际数据结构解析字段
            preparedStatement.setString(1, record);
            preparedStatement.addBatch();
        }

        int[] results = preparedStatement.executeBatch();
        //todo 需要验证是否已经持久化了
//        connection.commit(); // 这里先提交批次，最终事务由二阶段提交控制

        log.info("执行批量写入，影响行数: {}", results.length);

        // 清空当前批次
        records.clear();
        recordCount = 0;
        preparedStatement.clearBatch();
    }

    /**
     * 提交事务
     * 提交当前数据库连接中的所有未提交操作
     *
     * @throws SQLException SQL执行异常
     */

    public void commit() throws SQLException {
        // 提交最终事务
        if (!connection.getAutoCommit()) {
//            connection.commit();
        }
    }

    /**
     * 回滚事务
     * 回滚当前数据库连接中的所有未提交操作
     *
     * @throws SQLException SQL执行异常
     */

    public void rollback() throws SQLException {
        if (!connection.getAutoCommit()) {
            connection.rollback();
        }
    }

    /**
     * 关闭资源
     * 释放 PreparedStatement 和 Connection 资源
     */

    /**
     * 关闭连接也会提交事务
     */
    public void close() {
        try {
            if (preparedStatement != null && !preparedStatement.isClosed()) {
                preparedStatement.close();
            }
            if (connection != null && !connection.isClosed()) {
                connection.close();
            }
        } catch (SQLException e) {
            log.error("关闭连接失败", e);
        }
    }

    /**
     * 获取当前记录计数
     *
     * @return 当前缓冲区中的记录数量
     */

    public int getRecordCount() {
        return recordCount;
    }
    /**
     * 获取数据记录列表副本（用于状态恢复的序列化）
     *
     * @return 数据记录列表的副本
     */
    public List<String> getRecords() {
        return new ArrayList<>(records);
    }

    //序列化
    private void writeObject(ObjectOutputStream out) throws IOException {
        out.defaultWriteObject();
        // 只序列化可序列化的字段
    }
    private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
        in.defaultReadObject();
        // 反序列化后，transient 字段会自动设为 null
    }
}